package com.yjd.bi.etl;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.yjd.bi.canal.model.CanalMessageModel;
import com.yjd.bi.canal.model.Entrys;
import com.yjd.bi.canal.model.Records;
import com.yjd.bi.canal.service.ICanalMessageService;
import com.yjd.bi.report.model.DataChangeRecordModel;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.comm.rpm.RpmServiceKeyEnum;
import com.yjd.comm.util.FrameUtil;
import com.yjd.comm.util.ServiceHolder;
import com.yjd.comm.util.StringUtil;

public class Test {
	
	public static void main(String[]  args) throws Exception{
		ICanalMessageService canalMessageService=ServiceHolder.getService(ICanalMessageService.class, "canalMessageService", RpmServiceKeyEnum.RPM_BI_SERVICE_KEY_E);
		Map paramMap=FrameUtil.newHashMap("status",0,"event_type_min",EventType.INSERT.getNumber(),"event_type_max",EventType.DELETE.getNumber());
		List<CanalMessageModel> list=new ArrayList<CanalMessageModel>();
		list=canalMessageService.getModelList(paramMap,DbWREnums.WRITE);
		transfer(list,canalMessageService);
	}

	public static boolean transfer(List<CanalMessageModel> list,ICanalMessageService canalMessageService) throws Exception{
		List<DataChangeRecordModel> changelogs=FrameUtil.newArrayList();
		for(CanalMessageModel cm:list){
			List<Entrys> elist=JSON.parseArray(cm.getEntrydata(), Entrys.class);
			System.out.println(elist.size());
			switch (cm.getEvent_type()) {
				case EventType.INSERT_VALUE :
					{
						List<String> fieldList=new ArrayList<>();
						List<String> valueList=new ArrayList<>();
						for(Entrys e:elist){
							if(StringUtil.isNotEmpty(e.getValue_new())){
								fieldList.add(e.getField());
								valueList.add(e.getValue_new());
							}
						}
						System.out.println("insert start");
						if(fieldList.size()>0&&valueList.size()>0){
							canalMessageService.insertByCanalMessage(cm.getTable_name(), fieldList, valueList);
							System.out.println("inserting...");
						}
						System.out.println("insert end");
						break;
					}
				case EventType.UPDATE_VALUE :{
					Records keymap=new Records();
					List<Records> changeList=FrameUtil.newArrayList();
					for(Entrys e:elist){
						if(e.getIskey()!=null&&e.getIskey().intValue()==1){
							keymap.setField(e.getField());
							keymap.setValue(e.getValue_new());
						}
						if(e.getUpdate().intValue()==1){
							changeList.add(new Records(e.getField(),e.getValue_new()));
						}
					}
					System.out.println("UPDATE start");
					if(keymap!=null&&changeList.size()>0){
						canalMessageService.updateByCanalMessage(cm.getTable_name(), keymap, changeList);
						System.out.println("UPDATEing...");
					}
					System.out.println("UPDATE end");
					break;
				}
				case EventType.DELETE_VALUE:{
					Records keymap=new Records();
					for(Entrys e:elist){
						if(e.getIskey().intValue()==1){
							keymap.setField(e.getField());
							keymap.setValue(e.getValue_old());
						}
					}
					System.out.println("DELETE start");
					if(keymap!=null){
						canalMessageService.deleteByCanalMessage(cm.getTable_name(), keymap);
						System.out.println("DELETE ending...");
					}
					System.out.println("DELETE end");
				}
				default :
					break;
			}
		}
		return true;
	}
}
